package com.st.jms.kafka;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * 自动增长的int类型key
 */
public class ProducerAKClient {
  private Producer<Integer, String> producer;

  private Integer sendBatch;
  private List<KeyedMessage<Integer, String>> messages;

  public ProducerAKClient(String brokerList, Integer sendBatch) {
    super();
    this.sendBatch = sendBatch;
    this.messages = new ArrayList<>(sendBatch);
    Properties props = new Properties();
    props.put("metadata.broker.list", brokerList);
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("request.required.acks", "-1");
    ProducerConfig config = new ProducerConfig(props);
    this.producer = new Producer<>(config);
  }

  /**
   * 批量发送消息 将本地包含不同topic的message队列发送到kafka
   */
  private void sendBatch() {
    if (messages.size() >= this.sendBatch) {
      producer.send(messages);
      messages.clear();
    }
  }

  /**
   * 将指定topic的消息批量存储到本地
   *
   * @param topic    消息主题
   * @param contents 消息内容队列
   */
  public synchronized void putMessages(String topic, List<String> contents) {
    for (String content : contents) {
      KeyedMessage<Integer, String> message = new KeyedMessage<>(topic, content);
      messages.add(message);
    }
    this.sendBatch();
  }

  /**
   * 将指定topic的消息发送到kafka,忽略sendBath设置
   *
   * @param topic   消息主题
   * @param content 消息内容
   */
  public void sendMessage(String topic, String content) {
    KeyedMessage<Integer, String> message = new KeyedMessage<>(topic, content);
    producer.send(message);
  }

}
